package com.zuorenke.serialize;

import com.zuorenke.help.ClassUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.AbstractMessageConverter;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.utils.SerializationUtils;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.remoting.rmi.CodebaseAwareObjectInputStream;

import java.io.*;

/**
 * @author Z.R.K
 * @description
 * @create 2019-05-20 16:06:04
 **/
@Slf4j
public class DefaultMQMessageConverter  extends AbstractMessageConverter implements BeanClassLoaderAware {
	public static final String DEFAULT_CHARSET = "UTF-8";
	private static final String FAILED_CONVERT_TEXT = "failed to convert text-based Message content";
	private static final String FAILED_CONVERT_OBJECT = "failed to convert serialized Message content";
	private static final String FAILED_CREATE_TEXT = "failed to convert to Message content";
	private static final String FAILED_CREATE_OBJECT = "failed to convert to serialized Message content,check if the serialize and deserialize is the same serializer";
	
	private volatile String defaultCharset = DEFAULT_CHARSET;
	
	private String codebaseUrl;
	
	private ClassLoader beanClassLoader = ClassUtils.getClassLoader();
	
	private SerializableAdapter serializableAdapter = new DefaultSerializableAdapter();
	
	@Override
	public void setBeanClassLoader(ClassLoader classLoader) {
		this.beanClassLoader = classLoader;
	}
	
	public void setDefaultCharset(String defaultCharset) {
		this.defaultCharset = (defaultCharset != null) ? defaultCharset : DEFAULT_CHARSET;
	}
	
	public void setCodebaseUrl(String codebaseUrl) {
		this.codebaseUrl = codebaseUrl;
	}
	
	/*
	 * 反序列化：将amqp消息转换成java对象
	 *
	 * (non-Javadoc)
	 * @see org.springframework.amqp.support.converter.AbstractMessageConverter#fromMessage(org.springframework.amqp.core.Message)
	 */
	public Object fromMessage(Message message) {
		Object content = null;
		MessageProperties properties = message.getMessageProperties();
		if (properties != null) {
			String contentType = properties.getContentType();
			if (contentType != null && contentType.startsWith("text")) {
				String encoding = properties.getContentEncoding();
				if (encoding == null) {
					encoding = this.defaultCharset;
				}
				try {
					content = new String(message.getBody(), encoding);
				}
				catch (UnsupportedEncodingException e) {
					log.error(FAILED_CONVERT_TEXT, e);
					throw new MessageConversionException(
							FAILED_CONVERT_TEXT, e);
				}
			}
			else if (contentType != null &&
					contentType.equals(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT)) {
				try {
					if(null != serializableAdapter){
						content = serializableAdapter.deserialize(message.getBody());
					}else{
						content = SerializationUtils.deserialize(createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));
					}
				} catch (IOException |IllegalArgumentException e) {
					log.error(FAILED_CONVERT_OBJECT, e);
					throw new MessageConversionException(
							FAILED_CONVERT_OBJECT, e);
				}
			}
		}
		if (content == null) {
			content = message.getBody();
		}
		return content;
	}
	
	
	/*
	 * 序列化：将对象序列化成amqp消息
	 *
	 * (non-Javadoc)
	 * @see org.springframework.amqp.support.converter.AbstractMessageConverter#createMessage(java.lang.Object, org.springframework.amqp.core.MessageProperties)
	 */
	protected Message createMessage(Object object, MessageProperties messageProperties) {
		byte[] bytes = null;
		if (object instanceof byte[]) {
			bytes = (byte[]) object;
			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_BYTES);
		}
		else if (object instanceof String) {
			try {
				bytes = ((String) object).getBytes(this.defaultCharset);
			}
			catch (UnsupportedEncodingException e) {
				log.error(FAILED_CREATE_TEXT, e);
				throw new MessageConversionException(
						FAILED_CREATE_TEXT, e);
			}
			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
			messageProperties.setContentEncoding(this.defaultCharset);
		}
		else if (object instanceof Serializable) {
			try {
				if(null != serializableAdapter){
					bytes = serializableAdapter.serialize(object);
				}else{
					bytes = SerializationUtils.serialize(object);
				}
			} catch (IllegalArgumentException e) {
				log.error(FAILED_CREATE_OBJECT, e);
				throw new MessageConversionException(
						FAILED_CREATE_OBJECT, e);
			}
			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT);
		}
		if (bytes != null) {
			messageProperties.setContentLength(bytes.length);
		}
		return new Message(bytes, messageProperties);
	}
	
	protected ObjectInputStream createObjectInputStream(InputStream is, String codebaseUrl) throws IOException {
		return new CodebaseAwareObjectInputStream(is, this.beanClassLoader, codebaseUrl);
	}
}
